package com.glc.gao.gavin.rxjavademo.operator;

import android.view.View;

import com.glc.gao.gavin.rxjavademo.R;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;

public class MergeActivity extends BaseActivity {
    @Override
    public int getDescribeText() {
        return R.string.describe_merge;
    }

    @Override
    protected int getContextView() {
        return R.layout.activity_merge;
    }

    public void executeMerge(View view) {
        Observable.merge(Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Thread.sleep((long) (2000 + Math.random() * 2000));
                print("send：" + 1);
                e.onNext(1);
                Thread.sleep((long) (1 + Math.random() * 200));
                print("send：" + 2);
                e.onNext(2);
                Thread.sleep((long) (1 + Math.random() * 200));
                print("send：" + 3);
                e.onNext(3);
                Thread.sleep((long) (1 + Math.random() * 10));
            }
        }), Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                print("send：" + 4);
                e.onNext(4);
                Thread.sleep((long) (1 + Math.random() * 200));
                print("send：" + 5);
                e.onNext(5);
                Thread.sleep((long) (1 + Math.random() * 200));
                print("send：" + 6);
                e.onNext(6);
                Thread.sleep((long) (1 + Math.random() * 200));
            }
        })).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                print("接受：" + integer);
            }
        });
    }
}
